package com.hfzy.ihk.web.freeswitchCall.job.freeswitch;

import com.alibaba.fastjson.JSON;
import com.hfzy.ihk.common.util.string.StringUtil;
import com.hfzy.ihk.facade.freeswitchCall.service.FreeswitchCallUpdateCallHistoryService;
import com.hfzy.ihk.web.freeswitchCall.constant.RedisKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Profile;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Set;

/**
 * Created by YANFA on 2019/3/18.
 */
@Component
@Slf4j
@Profile({"pro","dev"})
public class PushDataToHbaseJob {

    private static Boolean isDataJobfinish = false;

    @Autowired
    private RedisTemplate redisTemplate;



    @Autowired(required = false)
    private FreeswitchCallUpdateCallHistoryService freeswitchCallUpdateCallHistoryService;

    /**
     * 不管是录音失败还是呼叫记录失败都全部重推，因为hbase那边是跟进sessionId更新的 ..补推hbase 了
     */
//    @Scheduled(cron = "0 0/5 * * * ?")
    public void rePushRecord(){
        if(isDataJobfinish)
            return;
        isDataJobfinish = false;
        try {
            //获取推送到es失败的通话记录的sessionId
            Set<String> pushFailedDataIds = redisTemplate.opsForSet().members(RedisKey.PUSH_RECORD_TO_HBASE_FAILED_SET);
            log.info(">>>>>>>>>>>>>>>FreeswitchCall执行补推数据到hbase redis 大小为：{}", pushFailedDataIds.size());

            if (pushFailedDataIds != null && !pushFailedDataIds.isEmpty()) {

                for (String sessionid : pushFailedDataIds) {
                    if (StringUtil.isEmpty(sessionid)) {
                        redisTemplate.opsForSet().remove(RedisKey.PUSH_TO_ES_FAILED_HISTORY, sessionid);
                    } else {
                        Map<String, String> callHistoryMap = redisTemplate.opsForHash().entries(RedisKey.CALL_HISTORY_RPIX + sessionid);
                        if (callHistoryMap == null || callHistoryMap.size() == 0) {
                            //假如为空,删除,免得堆积
                            redisTemplate.opsForSet().remove(RedisKey.PUSH_RECORD_TO_HBASE_FAILED_SET, sessionid); //这里会引起数据丢失，数据有效时间为3天
                        } else {
                            try {
                                String rs = freeswitchCallUpdateCallHistoryService.upsertCallHistory(JSON.toJSONString(callHistoryMap));
                                if ("1".equals(rs)) {
                                    redisTemplate.opsForSet().remove(RedisKey.PUSH_RECORD_TO_HBASE_FAILED_SET, sessionid); //推送成功，删除当前的key
                                }
                            } catch (Exception e) {
                                log.error(">>>>>>>推送hbase异常", e);
                            }
                        }
                    }
                }
            }
        }catch (Exception e){
            log.error(">>>>>>>>重新推送数据到hbase job 失败~",e);
        }
        isDataJobfinish = false;
    }
}
